package com.bxl.chooser.bean;

import com.bxl.netty.rpc.common.pool.ThreadPoolUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.PriorityQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * <>
 *
 * @author baixl
 * @date 2021/3/29
 */
public class MessageContainer {
    private Logger log = LoggerFactory.getLogger(MessageContainer.class);
    private MessageBean[] messages;
    private Map<String, MessageBean> map;
    private PriorityQueue<String> priorityQueue;
    /**
     * 日志索引-应该持久化，从本地读取》 commitLogIndex-logIndex 是可以提交但是未提交的数据
     */
    private int logIndex;
    /**
     * 提交的日志索引-应该持久化，从本地读取.commitLogIndex 以前的数据是都已经提交的不可修改的
     */
    private int commitLogIndex;

    public MessageContainer() {
        messages = new MessageBean[1024];
        map = new ConcurrentHashMap<>(1024);
        priorityQueue = new PriorityQueue<>(1024);
        logIndex = 0;
        commitLogIndex = -1;
        //打印数据，做数据核对
        ThreadPoolUtil.scheduleAtFixedRate(() -> {
            StringBuilder sb = new StringBuilder();
            for (int i = 0; i < logIndex; i++) {
                MessageBean msg = messages[i];
                sb.append(msg.getMessageId()).append(":").append(msg.getIndex()).append(":").append(msg.getContent())
                        .append(":").append(msg.getState().getDesc()).append("|");
            }
            System.out.println(sb.toString());

        }, 1, 1, TimeUnit.MINUTES);

    }


    public void commitLogIndexIncrease(AppInfo appInfo) {
        if (this.commitLogIndex < (this.logIndex - 1)) {
            this.commitLogIndex++;
            commitMessage(this.commitLogIndex,appInfo);
        }
    }

    /**
     *
     */
    public void refresh() {
        messages = new MessageBean[1024];
        map = new ConcurrentHashMap<>(1024);
        logIndex = 0;
        commitLogIndex = -1;
        //异步写入磁盘-或者多个缓存循环使用
    }

    public MessageBean getMessage(int index) {
        if (index >= logIndex) {
            return null;
        }
        return messages[index];
    }

    public MessageBean getMessage(String messageId) {
        return map.get(messageId);
    }


    /**
     * 需要写log文件
     *
     * @param message
     */
    public boolean addMessage(MessageBean message, AppInfo appInfo) {
        log.error("addMessage {}", logIndex);
        if (message == null) {
            return false;
        }
        //保证密等性
        MessageBean existsMessage = map.get(message.getMessageId());
        //消息已经存在-可能是客户端重试
        if (existsMessage != null) {
            log.error("消息已经存在 {}", message.getMessageId());
            return true;
        }
        message.setState(MessageBean.MessageState.UNCOMMIT);
        if (appInfo.getState() != AppInfo.AppState.LEADER) {
            if (message.getIndex() > logIndex) {
                log.error("{} 消息索引大于 logindex {}", message.getIndex(), logIndex);
                return false;
            } else if (message.getIndex() < logIndex) {
                int msgIndex = message.getIndex();
                //向后回滚,删除map中的数据
                while (logIndex > msgIndex) {
                    map.remove(messages[(logIndex - 1)].getMessageId());
                    logIndex--;
                }
                logIndex = message.getIndex();
            } else if (message.getIndex() == logIndex) {
                messages[logIndex] = message;
                map.put(message.getMessageId(), message);
                logIndex++;
                log.error("{} 消息添加成功 logindex {}", message.getIndex(), logIndex);
            } else {
                log.error("{} 消息索引不等于 logindex {}", message.getIndex(), logIndex);
                return false;
            }
        } else {
            message.setIndex(logIndex);
            messages[logIndex] = message;
            map.put(message.getMessageId(), message);
            log.error("leader 消息写入成功 索引 {}", logIndex);
            logIndex++;
        }
        if (logIndex >= 1024) {
            refresh();
        }
        return true;
    }

    public boolean rollback(String messageId) {
        if (StringUtils.isEmpty(messageId)) {
            return false;
        }
        MessageBean message = map.get(messageId);
        if (message == null) {
            log.error("{} not exists", messageId);
            return false;
        }
        if (message.getIndex() != (logIndex - 1)) {
            log.error("{} message rollback fail", messageId);
        } else {
            logIndex--;
        }
        return true;
    }

    /**
     * leader提交消息失败-重复尝试。直到提交成功
     *
     * @param commitIndex
     * @param appInfo
     * @return
     */
    public boolean commitMessage(int commitIndex, AppInfo appInfo) {
        if (commitIndex < 0) {
            return false;
        }
        if (appInfo.getState() != AppInfo.AppState.LEADER) {
            if (commitIndex < logIndex) {
                for (int i = this.commitLogIndex; i < commitIndex; i++) {
                    messages[i + 1].setState(MessageBean.MessageState.COMMITTED);
                }
                this.commitLogIndex = commitIndex;
            } else {
                return false;
            }
        } else {
            //leader 一条一条的提交。
            messages[commitIndex].setState(MessageBean.MessageState.COMMITTED);
        }
        return true;
    }

    /**
     * 需要写log文件
     *
     * @param messageId
     */
    public boolean deleteMessage(String messageId) {
        if (StringUtils.isEmpty(messageId)) {
            return false;
        }
        MessageBean message = map.get(messageId);
        if (message == null) {
            System.out.println(String.format("%s not exists", messageId));
            return false;
        }
        message.setState(MessageBean.MessageState.DELETED);
        return true;
    }

    public int getLogIndex() {
        return logIndex;
    }

    public void setLogIndex(int logIndex) {
        this.logIndex = logIndex;
    }

    public int getCommitLogIndex() {
        return commitLogIndex;
    }

    public void setCommitLogIndex(int commitLogIndex) {
        this.commitLogIndex = commitLogIndex;
    }
}
